package com.atguigu.edu.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.EduConfig;

import com.atguigu.edu.realtime.utils.DimUtil;
import com.atguigu.edu.realtime.utils.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.Collection;
import java.util.Set;

public class DimSinkFunction implements SinkFunction<JSONObject> {
    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {

        String tableName = jsonObj.getString("sink_table");
        jsonObj.remove("sink_table");

        String type = jsonObj.getString("type");
        jsonObj.remove("type");

        Set<String> keys = jsonObj.keySet();
        Collection<Object> values = jsonObj.values();
        String upsertSql = "upsert into " + EduConfig.PHOENIX_SCHEMA + "." + tableName
                + " (" + StringUtils.join(keys, ",") + ") " +
                "values('" + StringUtils.join(values, "','") + "')";
        System.out.println("向Phoenix表中插入upsertSql:" + upsertSql);

        PhoenixUtil.executesql(upsertSql);
        if("update".equals(type)){
            DimUtil.delCached(tableName,jsonObj.getString("id"));
        }




    }
}
